package net.bwie.realtime.jtp.dws.log.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;

import net.bwie.realtime.jtp.common.utils.DateTimeUtil;
import net.bwie.realtime.jtp.dws.log.bean.PageViewBean;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;

import java.util.HashMap;

/**
 * 将PageLog页面浏览日志数据封装到实体类PageViewBean对象
 *
 * @author xuanyu
 * @date 2025/5/19
 */
public class PageViewBeanMapFunction extends RichMapFunction<String, PageViewBean> {
    private transient ValueState<String> lastVisitDateState = null;

    @Override
    public void open(Configuration parameters) throws Exception {

        // todo 每个mid使用状态保存最近一次访问日期，由于统计每天UV，所以状态生命周期为1天即可，
        // 可以设置状态TTL为1day，节省内存使用
        // 状态描述符
        ValueStateDescriptor<String> valueStateDescriptor = new ValueStateDescriptor<>(
                "last-visit-date", String.class
        );

        // ttl配置
        StateTtlConfig ttlConfig = StateTtlConfig
                .newBuilder(Time.days(1))
                .updateTtlOnCreateAndWrite()
                .neverReturnExpired()
                .build();
        valueStateDescriptor.enableTimeToLive(ttlConfig);

        // 2初始化
        lastVisitDateState = getRuntimeContext().getState(valueStateDescriptor);
    }

    @Override
    public PageViewBean map(String value) throws Exception {
        // 获取数据
        JSONObject jsonObject = JSON.parseObject(value);
        if (jsonObject == null) {
            return null; // 或者记录日志并跳过
        }
        JSONObject object = jsonObject.getJSONObject("common");
        JSONObject pageObject = jsonObject.getJSONObject("page");
        String ar = object.getString("ar");
        //  获取地区字段表数据
        HashMap<String, String> areaDic = getAreaDic();
        //  获取省份
        String province = areaDic.get(ar);

        Long uvCount = 0L;
        String lastVisitDate = lastVisitDateState.value();
        String visitDate = DateTimeUtil.convertLongToString(jsonObject.getLong("ts"), DateTimeUtil.DATE_FORMAT);
        if (lastVisitDate == null || !lastVisitDate.equals(visitDate)) {
            uvCount = 1L;
            lastVisitDateState.update(visitDate);
        }

        return new PageViewBean(
                null,
                null,
                object.getString("ba"),
                object.getString("ch"),
                province,
                object.getString("is_new"),
                pageObject.get("last_page_id") == null ? 1L : 0L,
                1L,
                pageObject.getLong("during_time"),
                uvCount,
                jsonObject.getLong("ts")
        );
    }

    /**
     * 预先加载地区字段表数据，放到Map集合中
     */
    private HashMap<String, String> getAreaDic() {
        // map集合
        HashMap<String, String> map = new HashMap<>();
        // 添加数据
        map.put("110000", "北京");
        map.put("120000", "天津");
        map.put("140000", "山西");
        map.put("150000", "内蒙古");
        map.put("130000", "河北");
        map.put("310000", "上海");
        map.put("320000", "江苏");
        map.put("330000", "浙江");
        map.put("340000", "安徽");
        map.put("350000", "福建");
        map.put("360000", "江西");
        map.put("370000", "山东");
        map.put("710000", "台湾");
        map.put("230000", "黑龙江");
        map.put("220000", "吉林");
        map.put("210000", "辽宁");
        map.put("610000", "陕西");
        map.put("620000", "甘肃");
        map.put("630000", "青海");
        map.put("640000", "宁夏");
        map.put("650000", "新疆");
        map.put("410000", "河南");
        map.put("420000", "湖北");
        map.put("430000", "湖南");
        map.put("440000", "广东");
        map.put("450000", "广西");
        map.put("460000", "海南");
        map.put("810000", "香港");
        map.put("820000", "澳门");
        map.put("510000", "四川");
        map.put("520000", "贵州");
        map.put("530000", "云南");
        map.put("500000", "重庆");
        map.put("540000", "西藏");
        // 返回集合
        return map;
    }

    @Override
    public void close() throws Exception {
        // step4. 关闭
        super.close();
    }
}
